package top.go2do.deviceSync.config;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONString;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.jsontype.impl.LaissezFaireSubTypeValidator;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.Subscription;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import top.go2do.common.module.OperationLogBO;
import top.go2do.deviceSync.business.operationLog.service.OperationLogService;

import java.time.Duration;
import java.util.Map;

/**
 * @Description
 * @Author xingyj
 * @CreateTime 2025/5/10 11:54
 */
@ConditionalOnProperty(prefix = "custom.log",name = "mode",havingValue = "redis")
@Configuration
@Slf4j
public class RedisLogConfig {

    @Value("${custom.log.redis.streamKey:operationLog}")
    private String logStreamKey;

    @Value("${custom.log.redis.consumerGroup:logConsumerGroup1}")
    private String logConsumerGroup;

    @Value("${custom.log.redis.consumer:logConsumer1}")
    private String logConsumer;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private OperationLogService operationLogService;

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> streamMessageListenerContainer(
            RedisConnectionFactory redisConnectionFactory) {

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance,
                ObjectMapper.DefaultTyping.NON_FINAL);
        // 配置 JSON 序列化器
        Jackson2JsonRedisSerializer<OperationLogBO> logSerializer = new Jackson2JsonRedisSerializer<>(objectMapper, OperationLogBO.class);

        // 创建监听容器选项
        // 关键修正点：使用 MapRecord 类型
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, Object, Object>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .pollTimeout(Duration.ofSeconds(1))
                        .keySerializer(new StringRedisSerializer())      // 明确指定序列化器
                        .hashKeySerializer(new StringRedisSerializer())  // Hash key 序列化器
                        .hashValueSerializer(RedisSerializer.json())// Hash value 序列化器
                        .build();

        // 创建监听容器
        StreamMessageListenerContainer<String, MapRecord<String, Object, Object>> container =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

       initConsumerGroup(redisConnectionFactory);

        // 注册消费者
        container.receive(
                Consumer.from(logConsumerGroup, logConsumer),
                StreamOffset.create(logStreamKey, ReadOffset.lastConsumed()),
                message -> {
                    try {
                        RecordId recordId = message.getId();
                        log.info("receive logMessageId is {}", recordId);
                        // 1. 处理消息
                        Map<Object, Object> logMessage = message.getValue();
                        OperationLogBO logBO = BeanUtil.toBean(logMessage, OperationLogBO.class);
                        log.info("logBO:{}", logBO);
                        operationLogService.createLog(logBO);
                        // 2. 显式ACK确认
                        redisTemplate.opsForStream().acknowledge(logStreamKey, logConsumerGroup, message.getId());
                    } catch (Exception e) {
                        // 3. 异常处理：记录日志或将消息标记为失败
                        log.error("logMessage handle error:", e);
                    }
                }
        );

        container.start();
        return container;
    }

    void initConsumerGroup(RedisConnectionFactory redisConnectionFactory) {
        try (RedisConnection connection = redisConnectionFactory.getConnection()) {
            // 检查消费者组是否存在，不存在则创建（避免重复初始化）
            connection.xGroupCreate(
                    logStreamKey.getBytes(),
                    logConsumerGroup,
                    StreamOffset.fromStart(logStreamKey).getOffset(),
                    true // 如果 Stream 不存在则自动创建
            );
        } catch (Exception e) {
            // 消费者组已存在时会抛出异常，此处可忽略
            log.error("init consumer error", e);
        }
    }
}
